package com.zhao.demo.unbound.demo03_source.sample01_collection

import com.zhao.demo.Raytek
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.api.scala._

import scala.io.Source


/**
 * Description: 使用flink的无界流的API,从集合中读取数据,进行处理<br/>
 * Copyright (c) ，2020 ， 赵 <br/>
 * This program is protected by copyright laws. <br/>
 * Date： 2020/11/26 14:29
 * 场景:在北京西站各个关键的出入的场所,安放了一些红外体温测量仪,对往来的行人进行体温的实时的监测
 * 需求:从日志文件中读物红外体温测量仪的内置传感器所实时采集到的数据,经由flink实时流基础处理,给出结果
 *
 * @author 柒柒
 * @version : 1.0
 */

object ReadDataFromCollection {
  def main(args: Array[String]): Unit = {
    //1.执行环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    //2.读取文件中的数据,转换成集合
    val lst: List[String] = Source.fromFile("a_input/raytek/raytek.log")
      .getLines()
      .toList

    //3.将集合中的数据封装到DataStream中去
    val dataStream: DataStream[String] = env.fromCollection(lst)

    //4.对无界流数据进行迭代处理,并显示结果
    dataStream.map(perEle =>{
      //raytek_3,36.389,alice,1582641129,北京西站南广场-公交站
      val arr = perEle.split(",")
      val id = arr(0).trim
      val temperature = arr(1).trim.toDouble
      val name = arr(2).trim
      val timestamp = arr(3).trim.toLong
      val location = arr(4).trim
      Raytek(id,temperature,name,timestamp,location)
    }).print("红外体温测量仪->")

    //5.启动
    env.execute(this.getClass.getSimpleName)
  }
}

















